package com.yinhai.redis.cluster;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.commons.pool.impl.GenericObjectPool.Config;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

import com.yinhai.redis.cluster.event.ClusterEvent;
import com.yinhai.redis.cluster.event.Event;
import com.yinhai.redis.cluster.event.EventBus;
import com.yinhai.redis.cluster.event.EventListener;
import com.yinhai.redis.cluster.event.filter.EventTypeFilter;
import com.yinhai.redis.cluster.model.KeyRange;
import com.yinhai.redis.cluster.model.ShardHolder;
import com.yinhai.redis.cluster.model.ShardInfo;
import com.yinhai.redis.cluster.monitors.ExistsMonitor;
import com.yinhai.redis.cluster.monitors.LivesNodeMonitor;
import com.yinhai.redis.cluster.monitors.NodeExistsMonitor;
import com.yinhai.redis.cluster.serializer.ISerializable;
import com.yinhai.redis.cluster.serializer.KryoSerializer;
import com.yinhai.redis.cluster.sync.SyncThread;

/**
 * 缓存服务
 * @author dreamlee
 *
 */
public class ClusterService implements EventListener{
	
	/**
	 * 集群信息
	 */
	private List<ShardInfo> shards=Collections.synchronizedList(new ArrayList<ShardInfo>(16));
	
	/**
	 * 
	 */
	private List<JedisPool> pools=Collections.synchronizedList(new ArrayList<JedisPool>(16));
	
	/**
	 * 断开连接的节点
	 */
	private List<ShardInfo> disconnectedShards=Collections.synchronizedList(new ArrayList<ShardInfo>(16));
	
	public static final int HASH_BASE=(int)Math.pow(2, 20);
	
	private ZooKeeper zk;
	
	private CountDownLatch latch=new CountDownLatch(1);
	
	private List<KeyMatchers> matchers=new ArrayList<KeyMatchers>();
	
	private ISerializable serializer=new KryoSerializer();
	
	private ExecutorService executor=Executors.newCachedThreadPool();
	
	private HeartBeatThread heartBeat;
	
	/**
	 * 添加分片服务器
	 * @param shardinfo
	 */
	public void addShard(ShardInfo shardinfo){
		shards.add(shardinfo);
	}
	
	
	/**
	 * 初始化缓存集群
	 * @throws IOException
	 * @throws InterruptedException
	 * @throws KeeperException
	 */
	public void initialize() throws IOException, InterruptedException, KeeperException{
		EventBus.getInstance().addListener(new EventTypeFilter(ClusterEvent.class), this);
		
		Config config=new Config();
		config.maxActive=200;
		config.maxIdle=100;
		config.maxWait=2000;
		
		int perSize=HASH_BASE/shards.size();
		int offset=HASH_BASE % shards.size();
		int startIndex=0;
		
		List<ShardHolder> holders=new ArrayList<ShardHolder>();
		//初始化连接池
		for(Iterator<ShardInfo> iter=shards.iterator();iter.hasNext();){
			ShardInfo info=iter.next();
			try{
				JedisPool pool=new JedisPool(config, info.getMaster().getHost(), info.getMaster().getPort());
				pools.add(pool);
				
				//分配各个服务器的key范围
				KeyRange keyRange=new KeyRange(startIndex, startIndex+perSize-1);
				//备份节点的key范围
				if(shards.indexOf(info)!=shards.size()-1){
					info.setSlave(shards.get(shards.indexOf(info)+1));
					info.addKeyRange(keyRange);
				}else{
					info.setSlave(shards.get(0));
					keyRange.setEnd(keyRange.getEnd()+offset);
					info.addKeyRange(keyRange);
				}
				startIndex+=perSize;
				
				Jedis conn=pool.getResource();
				try{
					//对服务器进行一次ping,检测服务器是否存活。
					conn.ping();
				}finally{
					pool.returnResource(conn);
				}
				KeyMatchers matcher=new KeyMatchers();
				matcher.setInfo(info);
				matcher.setPool(pool);
				matchers.add(matcher);
				
				ShardHolder holder=new ShardHolder();
				holder.setPool(pool);
				holder.setInfo(info);
				holder.setMatcher(matcher);
				holders.add(holder);
			}catch(Exception e){
				iter.remove();
			}
		}
		
		
		//各节点之间环形互备
		for(int i=0;i<shards.size();i++){
			if(i==0){
				matchers.get(i).setMasterPool(pools.get(pools.size()-1));
			}else{
				matchers.get(i).setMasterPool(pools.get(i-1));
			}
		}
		
		//初始化zookeeper
		initZookeeper();
		if(zk.exists("/redis", false)==null){
			zk.create("/redis", null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
		}
		if(zk.exists("/redis/shards", false)==null){
			zk.create("/redis/shards", null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
		}
		
		if(zk.exists("/redis/lives", false)==null){
			zk.create("/redis/lives", null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
		}else{
			List<String> children=zk.getChildren("/redis/lives", false);
			for(String s:children){
				zk.delete("/redis/lives/"+s, 0);
			}
		}
		
		for(ShardInfo info:shards){
			if(zk.exists("/redis/shards/"+info.getName(), false)==null){
				zk.create("/redis/shards/"+info.getName(), serializer.serialize(info), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
			}
			zk.create("/redis/lives/"+info.getName(), null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
			
			//监听节点
			ExistsMonitor monitor=new NodeExistsMonitor();
			monitor.setShardInfo(info);
			monitor.setZk(zk);
			zk.exists("/redis/shards/"+info.getName(), monitor);
			
		}
		
		LivesNodeMonitor livesMonitor=new LivesNodeMonitor();
		livesMonitor.setZk(zk);
		livesMonitor.setHolders(holders);
		zk.getChildren("/redis/lives", livesMonitor);
		
		heartBeat=new HeartBeatThread();
		heartBeat.setZk(zk);
		heartBeat.setHolders(holders);
		heartBeat.setDaemon(true);
		heartBeat.start();
	}
	
	
	/**
	 * 初始化Zookeeper
	 * @throws IOException
	 * @throws InterruptedException
	 */
	private void initZookeeper() throws IOException, InterruptedException{
		zk=new ZooKeeper("localhost", 2000, new Watcher() {
			
			@Override
			public void process(WatchedEvent event) {
				if(event.getState()==KeeperState.SyncConnected){
					latch.countDown();
				}
			}
		});
		latch.await();
	}
	
	/**
	 * 设置缓存
	 * @param key
	 * @param value
	 */
	public void set(final String key,final Object value){
		final int hashcode=Math.abs(key.hashCode() % HASH_BASE);
		System.out.println(hashcode);
		for(KeyMatchers matcher:matchers){
			if(matcher.matchs(hashcode)){
				Jedis jedis=matcher.getPool().getResource();
				try{
					jedis.set(key.getBytes(), serializer.serialize(value));
					executor.submit(new Runnable() {
						
						@Override
						public void run() {
							for(KeyMatchers matcher:matchers){
								if(matcher.slaveMatchs(hashcode)){
									Jedis jedis=matcher.getPool().getResource();
									try{
										jedis.set(key.getBytes(), serializer.serialize(value));
									}finally{
										matcher.getPool().returnResource(jedis);
									}
								}
							}
						}
					});
					break;
				}finally{
					matcher.getPool().returnResource(jedis);
				}
			}
		}
	}
	
	
	public void del(final String key){
		final int hashcode=Math.abs(key.hashCode() % HASH_BASE);
		for(KeyMatchers matcher:matchers){
			if(matcher.matchs(hashcode)){
				Jedis jedis=matcher.getPool().getResource();
				try{
					jedis.del(key.getBytes());
					executor.submit(new Runnable() {
						
						@Override
						public void run() {
							for(KeyMatchers matcher:matchers){
								if(matcher.slaveMatchs(hashcode)){
									Jedis jedis=matcher.getPool().getResource();
									try{
										jedis.del(key.getBytes());
									}finally{
										matcher.getPool().returnResource(jedis);
									}
								}
							}
						}
					});
					break;
				}finally{
					matcher.getPool().returnResource(jedis);
				}
			}
		}
	}
	
	/**
	 * 根据key取得缓存中的值
	 * @param key
	 * @return
	 */
	public Object get(String key){
		int hashcode=key.hashCode() % HASH_BASE;
		for(KeyMatchers matcher:matchers){
			if(matcher.matchs(hashcode)){
				Jedis jedis=null;
				try{
					jedis=matcher.getPool().getResource();
					Object ret=serializer.deSerialize(jedis.get(key.getBytes()));
					if(ret==null){
						if(matcher.getMasterPool()!=null){
							Jedis master=matcher.getMasterPool().getResource();
							try{
								ret=serializer.deSerialize(master.get(key.getBytes()));
							}finally{
								matcher.getMasterPool().returnResource(master);
							}
						}
					}
					return ret;
				}catch(Exception e){
					if(matcher.getMasterPool()!=null){
						Jedis master=matcher.getMasterPool().getResource();
						try{
							return serializer.deSerialize(master.get(key.getBytes()));
						}finally{
							matcher.getMasterPool().returnResource(master);
						}
					}
				}finally{
					if(jedis!=null){
						matcher.getPool().returnResource(jedis);
					}
				}
			}
		}
		return null;
	}


	@Override
	public void handlerEvent(Event e) {
		ClusterEvent event=(ClusterEvent)e;
		switch(event.getType()){
		case ClusterEvent.SHARD_ADDED:
			//节点增加
			ShardInfo sinfo=(ShardInfo)e.getData();
			sinfo.setSlave(shards.get(0));
			shards.get(shards.size()-1).setSlave(sinfo);
			for(ShardInfo linfo:shards){
				if(linfo.getMasterRange().size()>1){
					sinfo.addKeyRange(linfo.getMasterRange().get(linfo.getMasterRange().size()-1));
				}
			}
			break;
		case ClusterEvent.SHARD_REMOVED:
			ShardHolder holder=(ShardHolder)e.getData();
			System.out.println(holder.getInfo().getName()+" shutdown");
			synchronized (matchers) {
				disconnectedShards.add(holder.getInfo());
				shards.remove(holder.getInfo());
			}
			int size=shards.size();
			for(KeyRange range:holder.getInfo().getMasterRange()){
				int start=range.getStart();
				int slice=(range.getEnd()-range.getStart()) / size;
				int offset=(range.getEnd()-range.getStart()) % size;
				for(ShardInfo info:shards){
					if(shards.indexOf(info)!=shards.size()-1){
						KeyRange newRange=new KeyRange(start, start+slice);
						info.addKeyRange(newRange);
						start+=slice;
					}else{
						KeyRange newRange=new KeyRange(start, start+slice+offset);
						info.addKeyRange(newRange);
					}
					try {
						zk.setData("/redis/shards/"+info.getName(), serializer.serialize(info), 0);
					} catch (KeeperException e1) {
						e1.printStackTrace();
					} catch (InterruptedException e1) {
						e1.printStackTrace();
					}
				}
			}
			holder.getInfo().getMasterShard().setSlave(holder.getInfo().getSlave());
			//开始将断开连接的节点上的数据分散到其它各节点
			SyncThread sync=new SyncThread();
			sync.setHolder(holder);
			sync.setMatchers(matchers);
			sync.setService(this);
			sync.setDaemon(true);
			sync.start();
			break;
		case ClusterEvent.SYNCHRONIZE_DONE:
			System.out.println("同步完成");
			break;
		case ClusterEvent.SYNCHRONIZE_ERROR:
			//TODO 节点间同步出错
			System.out.println("同步出错"+e.getData());
			break;
		}
	}
}
